#include <stdlib.h>
#include <string.h>
#include <stdio.h>
#include <errno.h>
#include <unistd.h>

#include "queue.h"
#include "rqueue.h"
#include "log.h"
#include "method.h"
#include "msg.h"

#define DEFAULT_RQUEUE_PORT 6018

static int rqueue_exit = 0;

pthread_mutex_t rexit = PTHREAD_MUTEX_INITIALIZER;
pthread_mutex_t lread = PTHREAD_MUTEX_INITIALIZER;

static queue_thread *th_pool;

int  make_rpool();
void drop_rpool();

int rqueue_init();

int rqueue_fini();

void lock_rqueue_main();

void lock_rqueue_main()
{
again:
	if(0 != pthread_mutex_lock(&rexit))
	{
		if(errno == EINTR) goto again;
		fprintf(stderr, "Call pthread_mutex_lock wexit failure!\n");
	}
}

int rqueue_init()
{
	//if(-1 == msg_init()) return -1;

	return 1;
}

int rqueue_fini()
{
	//if(-1 == msg_fini()) return -1;

	return 1;
}

int make_rpool()
{
	int i;
	th_pool = (queue_thread*)malloc(HANDLER_NUMBER*sizeof(queue_thread));
	if(th_pool == NULL)
	{
		fprintf(stderr, "call malloc th_pool error!\n");
		return -1;
	}

	for(i = 0; i < HANDLER_NUMBER; i++)
	{
		if(0 != pthread_create(&th_pool[i].tid, NULL, (void*)main_rqueue, (void*)i))
		{
			fprintf(stderr, "call pthread_create failure!\n");
			drop_rpool();
			return -2;
		}
	}

	return 1;
}

void drop_rpool()
{
	free(th_pool);
}

void rqueue_start()
{
	pthread_t th;
	rqueue_exit = 0;
	if(0 != pthread_create(&th, NULL, (void*)lock_rqueue_main, NULL))
	{
		fprintf(stderr, "call lock_rqueue_main failure!\n");
		return;
	}
	pthread_join(th,NULL);

	if(rqueue_init() < 0) return;
	if(make_rpool () < 0) return;

again:
	if(0 != pthread_mutex_lock(&rexit))
	{
		if(errno == EINTR) goto again;
		fprintf(stderr, "call pthread_mutex_lock rexit failed!\n");
		return;
	}

	drop_rpool();

#ifdef DEBUG
	fprintf(stderr, "rqueue end!\n");
#endif
}

void rqueue_stop()
{
	int i, count;
	rqueue_exit = 1;
	for(i = 0; i < HANDLER_NUMBER; i++)
	{
		if(0 != rqueue_read_unlock(i))
		{
			logger(PRIORITY_ERROR, "unlock read rqueue error at %d in '%s'.",
				__LINE__, __FILE__);
			continue;
		}

#ifdef DEBUG
		fprintf(stderr, "waiting for rqueue index: %d", i);
#endif
		count = 0;
		if(is_rqueue_lock[i] == 1)
			rqueue_read_unlock(i);
		if(rqueue_lock[i] == 1)
			rqueue_read_continue(i);
		while(th_pool[i].runing)
		{
#ifdef DEBUG
			fprintf(stderr, ".");
#endif
			usleep(50000);
			count++;
			if(count == 60) {
#ifdef DEBUG
				fprintf(stderr, " not exit in 3s.");
#endif
				break;
			}
		}
#ifdef DEBUG
		fprintf(stderr, "\n");
#endif
	}

	rqueue_fini();
	if(0 != pthread_mutex_unlock(&rexit))
	{
		logger(PRIORITY_ERROR, "Unlock pthread mutex error (%s) at %d in '%s'.",
			strerror(errno) , __LINE__, __FILE__);
	}
}

void main_rqueue(void* i)
{
	int ret, count = 0;
	void *data = NULL;
	int index = (int)i;
	queue_header header;
	th_pool[index].runing = 1;
	rqueue_lock[index] = 0;
	if(Q_EXEC_OK != rqueue_open(index)) {
		logger(PRIORITY_ERROR, "NO.%d rqueue open error at %d in '%s'.", index, __LINE__, __FILE__);
		goto on_terminate;
	}

start_rq_initialize:
	if(-1 == rq_initialize(index)) {
		logger(PRIORITY_ERROR, "NO.%d rqueue initialize error at %d in '%s'.", index, __LINE__, __FILE__);
		rqueue_lock[index] = 1;
	}
	
start_get_rqueue:
	if(rqueue_exit) {
#ifdef DEBUG
		logger(PRIORITY_DEBUG, "rqueue No.%d exit.", index);
#endif
		goto on_terminate;
	}

	if(rqueue_lock[index]) {
		/* the queue is user locked or some error occured. */
		logger(PRIORITY_DEBUG, "rqueue No.%d paused, it may user locked or some error occured.", index);
		rqueue_read_pause(index);

		if(rqueue_exit) goto on_terminate;

		/* reconfig the rqueue, when restart. */
		rq_destroy(index);
		goto start_rq_initialize;
	}

	while(0 < (ret = next(index, &header, &data))) {
		header.start_sign = status[index]->position;
		logger(PRIORITY_INFO, "No.%d header [%X, %d, %s, %d]\n%.116s\n",
			index, header.start_sign, header.zero, header.ext, header.size, (char*)data);

		if(data == NULL) {
			/* get no data, it must be at end of queue. */
			logger(PRIORITY_DEBUG, "get no data, it must be at end of queue.");
			break;
		}

		/* main handle function of rqueue, see method.c for detail. */
		while(0 > rq_handle(index, &header, data)) {
			logger(PRIORITY_ERROR, "error handler on position %d, retry count %d, at %d in '%s'",
				status[index]->position, count, __LINE__, __FILE__);

			usleep(1000000);
			if(count++ < 3)
				continue;

			logger(PRIORITY_ERROR, "retry more than 3 times, No.%d rqueue locked.", index);
			count = 0;
			free(data);
			data = NULL;
			rqueue_lock[index] = 1;
			goto start_get_rqueue;
		}
		free(data);
		data = NULL;

		if(-1 == forward(index)) {
			logger(PRIORITY_ERROR, "No.%d Forward error.", index);
			goto on_terminate;
		}

		if(1 == rqueue_exit) {
			logger(PRIORITY_DEBUG, "No.%d exit while forward at position %d.", index, status[index]->position);
			goto on_terminate;
		}

		if(rqueue_lock[index])
			goto start_get_rqueue;
	}

	if(ret == -1) {
		logger(PRIORITY_ERROR, "No.%d error while read queue at %d.\n", index, status[index]->position);
		goto on_terminate;
	}
	
	th_pool[index].count++;

	/* the lock will be opened by wqueue */
	if(-1 == rqueue_read_lock(index)) {
		logger(PRIORITY_ERROR, "%d rqueue read lock failure...", index);
		goto on_terminate;
	}

	goto start_get_rqueue;

on_terminate:
	th_pool[index].runing = 0;
	rq_destroy(index);
	rqueue_close(index);
#ifdef DEBUG
	logger(PRIORITY_DEBUG, "rqueue %d terminated.", index);
#endif
}
